/*  
 * @(#) KafkaConsumer.java Create on 2015年3月11日 下午5:25:20   
 *   
 * Copyright 2015 hiveview.
 */


package com.hiveview.kafka;

import java.util.Enumeration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.ResourceBundle;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import scala.inline;

/**
 * @KafkaConsumer.java
 * @created at 2015年3月11日 下午5:25:20 by zhanghongliang@hiveview.com
 *
 * @desc
 *
 * @author  zhanghongliang@hiveview.com
 * @version $Revision$
 * @update: $Date$
 */
public class KafkaConsumer {
	private final ConsumerConnector consumer;
	private final String charset = "utf-8";
	
	private final Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);
	private ExecutorService threadPool;
	
	public static final ResourceBundle Kafka_consumer = ResourceBundle.getBundle("kafka-consumer");
	private KafkaConsumer() {
		consumer = Consumer.createJavaConsumerConnector(createConsumerConfig());
		threadPool = Executors.newCachedThreadPool();
	}
	
	private static final class BUILDER{
		private static final KafkaConsumer INSTALL = new KafkaConsumer();
	}
	
	public static KafkaConsumer getInstall(){
		return BUILDER.INSTALL;
	}
	
	private ConsumerConfig createConsumerConfig() {
		Properties props = new Properties();
//		props.put("zookeeper.connect", KafkaProperties.zkConnect);
//		String groupid = "";
//		props.put("group.id",groupid);
//		props.put("zookeeper.session.timeout.ms", "10000");
//		props.put("zookeeper.sync.time.ms", "200");
//		props.put("auto.commit.interval.ms", "1000");

		Enumeration<String> enumprops = Kafka_consumer.getKeys();
		while(enumprops .hasMoreElements()){
			String key = enumprops .nextElement();
			props.put(key, Kafka_consumer.getObject(key));
		}
		
		return new ConsumerConfig(props);
	}
	
	/**
	 * 注册消息监听
	 * @Title: registerMsgListener
	 * @data:2015年3月11日下午5:58:22
	 * @author:zhanghongliang@hiveview.com
	 *
	 * @param topic
	 * @param listener
	 */
	public void registerMsgListener (String topic,Messagelistener listener){
		registerMsgListener(topic, 1, listener);
	}
	
	/**
	 * 注册消息监听
	 * @Title: registerMsgListener
	 * @data:2015年3月11日下午5:58:30
	 * @author:zhanghongliang@hiveview.com
	 *
	 * @param topic
	 * @param partitionsNum   有几个分区
	 * @param listener
	 */
	public void registerMsgListener(String topic,Integer partitionsNum,Messagelistener listener){
		if(listener==null){
			logger.error("listener 不能为null");
			return;
		}
		Map<String, Integer> topics = new HashMap<String, Integer>();
		topics.put(topic, partitionsNum);
		Map<String, List<KafkaStream<byte[], byte[]>>> streams = consumer.createMessageStreams(topics);
		List<KafkaStream<byte[], byte[]>> partitions = streams.get(topic);
		
		for (KafkaStream<byte[], byte[]> kafkaStream : partitions) {
			threadPool.execute(new MessageRunner(kafkaStream , listener));
		}
	}
	
	class MessageRunner implements Runnable {
		private KafkaStream<byte[], byte[]> partition;
		private Messagelistener messagelistener;
		String msg = null;
		MessageRunner(KafkaStream<byte[], byte[]> partition,Messagelistener messagelistener) {
			this.partition = partition;
			this.messagelistener = messagelistener;
		}
		@Override
		public void run() {
			ConsumerIterator<byte[], byte[]> it = partition.iterator();
			while (it.hasNext()) {
				// connector.commitOffsets();手动提交offset,当autocommit.enable=false时使用
				MessageAndMetadata<byte[], byte[]> item = it.next();
				try{
					msg = new String(item.message(),charset);
					logger.debug("接受到消息:{}",msg);
					System.out.println("partition:" +item.partition() + "|" + "offset:" + item.offset());
					messagelistener.rereceiveMsg(msg);// UTF-8,注意异常
				}catch(Exception e){
					logger.error("receivemsg error",e);
				}
			}
		}
	}
}
